package com.tc.vms.consumer;

import com.tc.vms.SLA;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.iface.IContext;
import com.tc.vms.iface.ISubscriber;
import com.tc.vms.iface.IVMSCallback;
import com.tc.vms.log.VmsClientLogger;
import com.tc.vms.utils.Utils;
import com.tc.vms.utils.Utils;
import org.slf4j.Logger;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

/**
 * Created by yonghua.zhang on 2016/1/12.
 */
public abstract class AbstractConsumer implements ISubscriber {
    protected final static Logger LOGGER = VmsClientLogger.getLog();
    // modifying the param of RECONSUME_DELAY_LEVEL is forbidden, for the corresponding retry-queue have exists.
    protected static final String RECONSUME_DELAY_LEVEL = "5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h";
    protected volatile boolean isOpen = false;
    protected volatile boolean isSuspend = false;
    protected int consumingThreadNum = Utils.getCpuCoreNum() * 2;

    protected final VmsConsumerProxy vmsConsumerProxy;
    protected volatile IVMSCallback callback;
    protected final QueueMetaData queueMetaData;
    protected final String instName;
    protected final String groupName;
    protected final SubQoS subQoS;
    protected final SLA sla;
    protected Map<Integer, Long> delayLevelTable = Collections.synchronizedMap(new HashMap<Integer, Long>());

    protected AbstractConsumer(VmsConsumerProxy vmsConsumerProxy) {
        this.vmsConsumerProxy = vmsConsumerProxy;
        queueMetaData = vmsConsumerProxy.getVmsMetaData();
        instName = queueMetaData.getConsumerInstName();
        groupName = queueMetaData.getRealQueueNmOnBroker();
        subQoS = vmsConsumerProxy.qos();
        sla = vmsConsumerProxy.sla();
    }

    @Override
    public synchronized void registerListener(IVMSCallback callback) {
        this.callback = callback;
    }

    @Override
    public void suspend() {
        isSuspend = true;
    }

    @Override
    public void resume() {
        isSuspend = false;
    }

    @Override
    public void setConsumingThreadNum(int threadNum) {
        this.consumingThreadNum = threadNum;
    }

    @Override
    public boolean init(IContext context) throws VmsClientException {
        return false;
    }

    @Override
    public boolean start() throws VmsClientException {
        return false;
    }

    @Override
    public void shutdown() {

    }

    @Override
    public SLA slaOption() {
        return null;
    }

    @Override
    public SubQoS qosOption() {
        return null;
    }

    protected boolean parseDelayLevel() {
        HashMap<String, Long> timeUnitTable = new HashMap<String, Long>();
        timeUnitTable.put("s", 1000L);
        timeUnitTable.put("m", 1000L * 60);
        timeUnitTable.put("h", 1000L * 60 * 60);
        timeUnitTable.put("d", 1000L * 60 * 60 * 24);
        try {
            String[] levelArray = RECONSUME_DELAY_LEVEL.split(" ");
            for (int i = 0; i < levelArray.length; i++) {
                String value = levelArray[i];
                String ch = value.substring(value.length() - 1);
                Long tu = timeUnitTable.get(ch);

                int level = i + 1;
                long num = Long.parseLong(value.substring(0, value.length() - 1));
                long delayTimeMillis = tu * num;
                this.delayLevelTable.put(level, delayTimeMillis);
            }
        } catch (Exception e) {
            LOGGER.error(String.format("parseDelayLevel levelString String %s error", RECONSUME_DELAY_LEVEL), e);
            return false;
        }
        return true;
    }

}
